#ifndef SSDB_IMPL_H_
#define SSDB_IMPL_H_

#include "binlog.h"
#include "iterator.h"
#include "leveldb/db.h"
#include "leveldb/slice.h"
#include "log.h"
#include "ssdb.h"
#include "t_hash.h"
#include "t_kv.h"
#include "t_queue.h"
#include "t_zset.h"

namespace ssdb {

inline static leveldb::Slice slice(const Bytes &b) {
  return leveldb::Slice(b.data(), b.size());
}

class SSDBImpl : public SSDB {
private:
  friend class SSDB;
  leveldb::DB *ldb;
  leveldb::Options options;

public:
  std::unique_ptr<BinlogQueue> binlogs;
  
  SSDBImpl();
  ~SSDBImpl() override;

  int flushdb() override;

  // return (start, end], not include start
  std::shared_ptr<Iterator> iterator(const std::string &start, const std::string &end,
                             uint64_t limit) override;
  std::shared_ptr<Iterator> rev_iterator(const std::string &start,
                                 const std::string &end, uint64_t limit) override;

  // void flushdb();
  uint64_t size() override;
  std::vector<std::string> info() override;
  void compact() override;
  int key_range(std::vector<std::string> *keys) override;

  /* raw operates */

  // repl: whether to sync this operation to slaves
  int raw_set(const Bytes &key, const Bytes &val) override;
  int raw_del(const Bytes &key) override;
  int raw_get(const Bytes &key, std::string *val) override;

  /* key value */

  int set(const Bytes &key, const Bytes &val,
                  char log_type = BinlogType::SYNC) override;
  int setnx(const Bytes &key, const Bytes &val,
                    char log_type = BinlogType::SYNC) override;
  int del(const Bytes &key, char log_type = BinlogType::SYNC) override;
  // -1: error, 1: ok, 0: value is not an integer or out of range
  int incr(const Bytes &key, int64_t by, int64_t *new_val,
                   char log_type = BinlogType::SYNC) override;
  int multi_set(const std::vector<Bytes> &kvs, int offset = 0,
                        char log_type = BinlogType::SYNC) override;
  int multi_del(const std::vector<Bytes> &keys, int offset = 0,
                        char log_type = BinlogType::SYNC) override;
  int setbit(const Bytes &key, int bitoffset, int on,
                     char log_type = BinlogType::SYNC) override;
  int getbit(const Bytes &key, int bitoffset) override;

  int get(const Bytes &key, std::string *val) override;
  int getset(const Bytes &key, std::string *val, const Bytes &newval,
                     char log_type = BinlogType::SYNC) override;
  // return (start, end]
  std::shared_ptr<KIterator> scan(const Bytes &start, const Bytes &end, uint64_t limit) override;
  std::shared_ptr<KIterator> rscan(const Bytes &start, const Bytes &end,
                           uint64_t limit) override;

  /* hash */

  int hset(const Bytes &name, const Bytes &key, const Bytes &val,
                   char log_type = BinlogType::SYNC) override;
  int hdel(const Bytes &name, const Bytes &key,
                   char log_type = BinlogType::SYNC) override;
  // -1: error, 1: ok, 0: value is not an integer or out of range
  int hincr(const Bytes &name, const Bytes &key, int64_t by,
                    int64_t *new_val, char log_type = BinlogType::SYNC) override;
  // int multi_hset(const Bytes &name, const std::vector<Bytes> &kvs, int
  // offset=0, char log_type=BinlogType::SYNC); int multi_hdel(const Bytes &name,
  // const std::vector<Bytes> &keys, int offset=0, char
  // log_type=BinlogType::SYNC);

  int64_t hsize(const Bytes &name) override;
  int64_t hclear(const Bytes &name) override;
  int hget(const Bytes &name, const Bytes &key, std::string *val) override;
  int hlist(const Bytes &name_s, const Bytes &name_e, uint64_t limit,
                    std::vector<std::string> *list) override;
  int hrlist(const Bytes &name_s, const Bytes &name_e, uint64_t limit,
                     std::vector<std::string> *list) override;
  std::shared_ptr<HIterator> hscan(const Bytes &name, const Bytes &start,
                           const Bytes &end, uint64_t limit) override;
  std::shared_ptr<HIterator> hrscan(const Bytes &name, const Bytes &start,
                            const Bytes &end, uint64_t limit) override;

  /* zset */

  int zset(const Bytes &name, const Bytes &key, const Bytes &score,
                   char log_type = BinlogType::SYNC) override;
  int zdel(const Bytes &name, const Bytes &key,
                   char log_type = BinlogType::SYNC) override;
  // -1: error, 1: ok, 0: value is not an integer or out of range
  int zincr(const Bytes &name, const Bytes &key, int64_t by,
                    int64_t *new_val, char log_type = BinlogType::SYNC) override;
  // int multi_zset(const Bytes &name, const std::vector<Bytes> &kvs, int
  // offset=0, char log_type=BinlogType::SYNC); int multi_zdel(const Bytes &name,
  // const std::vector<Bytes> &keys, int offset=0, char
  // log_type=BinlogType::SYNC);

  int64_t zsize(const Bytes &name) override;
  /**
   * @return -1: error; 0: not found; 1: found
   */
  int zget(const Bytes &name, const Bytes &key, std::string *score) override;
  int64_t zrank(const Bytes &name, const Bytes &key) override;
  int64_t zrrank(const Bytes &name, const Bytes &key) override;
  std::shared_ptr<ZIterator> zrange(const Bytes &name, uint64_t offset, uint64_t limit) override;
  std::shared_ptr<ZIterator> zrrange(const Bytes &name, uint64_t offset,
                             uint64_t limit) override;
  /**
   * scan by score, but won't return @key if key.score=score_start.
   * return (score_start, score_end]
   */
  std::shared_ptr<ZIterator> zscan(const Bytes &name, const Bytes &key,
                           const Bytes &score_start, const Bytes &score_end,
                           uint64_t limit) override;
  std::shared_ptr<ZIterator> zrscan(const Bytes &name, const Bytes &key,
                            const Bytes &score_start, const Bytes &score_end,
                            uint64_t limit) override;
  int zlist(const Bytes &name_s, const Bytes &name_e, uint64_t limit,
                    std::vector<std::string> *list) override;
  int zrlist(const Bytes &name_s, const Bytes &name_e, uint64_t limit,
                     std::vector<std::string> *list) override;
  int64_t zfix(const Bytes &name) override;

  int64_t qsize(const Bytes &name) override;
  // @return 0: empty queue, 1: item peeked, -1: error
  int qfront(const Bytes &name, std::string *item) override;
  // @return 0: empty queue, 1: item peeked, -1: error
  int qback(const Bytes &name, std::string *item) override;
  // @return -1: error, other: the new length of the queue
  int64_t qpush_front(const Bytes &name, const Bytes &item,
                              char log_type = BinlogType::SYNC) override;
  int64_t qpush_back(const Bytes &name, const Bytes &item,
                             char log_type = BinlogType::SYNC) override;
  // @return 0: empty queue, 1: item popped, -1: error
  int qpop_front(const Bytes &name, std::string *item,
                         char log_type = BinlogType::SYNC) override;
  int qpop_back(const Bytes &name, std::string *item,
                        char log_type = BinlogType::SYNC) override;
  int qfix(const Bytes &name) override;
  int qlist(const Bytes &name_s, const Bytes &name_e, uint64_t limit,
                    std::vector<std::string> *list) override;
  int qrlist(const Bytes &name_s, const Bytes &name_e, uint64_t limit,
                     std::vector<std::string> *list) override;
  int qslice(const Bytes &name, int64_t offset, int64_t limit,
                     std::vector<std::string> *list) override;
  int qget(const Bytes &name, int64_t index, std::string *item) override;
  int qset(const Bytes &name, int64_t index, const Bytes &item,
                   char log_type = BinlogType::SYNC) override;
  int qset_by_seq(const Bytes &name, uint64_t seq, const Bytes &item,
                          char log_type = BinlogType::SYNC) override;

private:
  int64_t _qpush(const Bytes &name, const Bytes &item,
                 uint64_t front_or_back_seq, char log_type = BinlogType::SYNC);
  int _qpop(const Bytes &name, std::string *item, uint64_t front_or_back_seq,
            char log_type = BinlogType::SYNC);
};

}
#endif
